local utils = import '../utils.libsonnet';
local zookeeper = import '../zookeeper/index.libsonnet';

{
  Kafka(name, _values)::
    local values = std.mergePatch(
      (import 'values.libsonnet'),
      _values,
    );

    local args = if values.size > 1 then
      [
        {
          id: std.toString(id),
          name: name + values.infix + std.toString(id),
        }
        for id in std.range(1, values.size)
      ]
    else
      [
        {
          id: '1',
          name: name,
        },
      ];

    local zookeeperAddress =
      if values.zookeeper.size > 1 then
        std.join(',', [
          name + '-zk-' + std.toString(id) + ':2181'
          for id in std.range(1, values.zookeeper.size)
        ])
      else
        name + '-zk:2181';

    zookeeper.Zookeeper(
      name + '-zk',
      values.zookeeper { infix: '-' },
    ) + [
      utils.Service(
        arg.name,
        std.mergePatch(values.service, {
          ports: {
            tcp: [9092, 9999],
          },
        })
      )
      for arg in args
    ] +
    [
      utils.PersistentVolumeClaim(
        arg.name + '-data',
        values.storage,
      )
      for arg in args
    ] +
    [
      utils.StatefulSet(
        arg.name,
        std.mergePatch(
          values,
          {
            env: utils.Envs(values.env {
              KAFKACFG_broker__id: std.toString(arg.id),
              KAFKACFG_zookeeper__connect: zookeeperAddress,
              KAFKAAUTOCFG_PORT: '9092',
              KAFKAAUTOCFG_ADVERTISED_HOST: arg.name,
            }),
            podLabels: {
              service: name,
            },
            volumes: [
              {
                name: 'vol-data',
                persistentVolumeClaim: {
                  claimName: arg.name + '-data',
                },
              },
            ],
            volumeMounts: [
              {
                name: 'vol-data',
                mountPath: '/data',
              },
            ],
          }
        )
      )
      for arg in args
    ],
}
